#include "in.h"

/*
MESSAGE HEAD PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|...|
|'0'|"ABCD"|...|
*/

int DispatchProcess( struct TcpdaemonServerEnvironment *p_env , struct MQAcceptedSession *p_accepted_session )
{
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	
	int			nret = 0 ;
	
	ResetAppendBufferParsePtr( p_accepted_session->comm_data );
	
	MEMUNPACK_COMM_VERSION( return -1 , p_accepted_session->comm_data , comm_version )
	if( comm_version != MQ_COMM_PROTO_VERSION_1 )
	{
		ERRORLOG( "invalid comm_version[%c][0x%02X]" , comm_version , comm_version )
		return TCPMAIN_RETURN_CLOSE;
	}
	
	MEMUNPACK_PROCESS_CODE( return -1 , p_accepted_session->comm_data , process_code )
	if( process_code.u32 == *(uint32_t*)(MQ_PROCESSCODE_CREATESTORE_REQUEST) )
	{
		nret = app_CreateStore( p_env , p_accepted_session ) ;
		if( nret == TCPMAIN_RETURN_CLOSE )
		{
			ERRORLOG( "app_CreateStore return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
		else
		{
			INFOLOG( "app_CreateStore return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
	}
	else if( process_code.u32 == *(uint32_t*)(MQ_PROCESSCODE_DESTROYSTORE_REQUEST) )
	{
		nret = app_DestroyStore( p_env , p_accepted_session ) ;
		if( nret == TCPMAIN_RETURN_CLOSE )
		{
			ERRORLOG( "app_DestroyStore return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
		else
		{
			INFOLOG( "app_DestroyStore return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
	}
	else if( process_code.u32 == *(uint32_t*)(MQ_PROCESSCODE_CREATETOPIC_REQUEST) )
	{
		nret = app_CreateTopic( p_env , p_accepted_session ) ;
		if( nret == TCPMAIN_RETURN_CLOSE )
		{
			ERRORLOG( "app_CreateTopic return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
		else
		{
			INFOLOG( "app_CreateTopic return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
	}
	else if( process_code.u32 == *(uint32_t*)(MQ_PROCESSCODE_DESTROYTOPIC_REQUEST) )
	{
		nret = app_DestroyTopic( p_env , p_accepted_session ) ;
		if( nret == TCPMAIN_RETURN_CLOSE )
		{
			ERRORLOG( "app_DestroyTopic return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
		else
		{
			INFOLOG( "app_DestroyTopic return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
	}
	else if( process_code.u32 == *(uint32_t*)(MQ_PROCESSCODE_PRODUCEMESSAGE_REQUEST) )
	{
		nret = app_ProduceMessage( p_env , p_accepted_session ) ;
		if( nret == TCPMAIN_RETURN_CLOSE )
		{
			ERRORLOG( "app_ProduceMessage return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
		else
		{
			INFOLOG( "app_ProduceMessage return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
	}
	else if( process_code.u32 == *(uint32_t*)(MQ_PROCESSCODE_SUBSCRIBEMESSAGE_REQUEST) )
	{
		nret = app_SubscribeMessage( p_env , p_accepted_session ) ;
		if( nret == TCPMAIN_RETURN_CLOSE )
		{
			ERRORLOG( "app_SubscribeMessage return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
		else
		{
			INFOLOG( "app_SubscribeMessage return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
	}
	else if( process_code.u32 == *(uint32_t*)(MQ_PROCESSCODE_CONSUMEMESSAGE_RESPONSE) )
	{
		nret = app_ConsumeMessage( p_env , p_accepted_session ) ;
		if( nret == TCPMAIN_RETURN_CLOSE )
		{
			ERRORLOG( "app_ConsumeMessage return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
		else
		{
			INFOLOG( "app_ConsumeMessage return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
	}
	else if( process_code.u32 == *(uint32_t*)(MQ_PROCESSCODE_PRODUCEDELAYMESSAGE_REQUEST) )
	{
		nret = app_ProduceDelayMessage( p_env , p_accepted_session ) ;
		if( nret == TCPMAIN_RETURN_CLOSE )
		{
			ERRORLOG( "app_ProduceDelayMessage return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
		else
		{
			INFOLOG( "app_ProduceDelayMessage return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
	}
	else if( process_code.u32 == *(uint32_t*)(MQ_PROCESSCODE_HEARTBEAT_RESPONSE) )
	{
		nret = app_HeartbeatResponse( p_env , p_accepted_session ) ;
		if( nret == TCPMAIN_RETURN_CLOSE )
		{
			ERRORLOG( "app_HeartbeatResponse return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
		else
		{
			INFOLOG( "app_HeartbeatResponse return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
	}
	else if( process_code.u32 == *(uint32_t*)(MQ_PROCESSCODE_PRODUCEDELAYMESSAGE_REQUEST) )
	{
		nret = app_ProduceDelayMessage( p_env , p_accepted_session ) ;
		if( nret == TCPMAIN_RETURN_CLOSE )
		{
			ERRORLOG( "app_ProduceDelayMessage return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
		else
		{
			INFOLOG( "app_ProduceDelayMessage return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
	}
	else if( process_code.u32 == *(uint32_t*)(MQ_PROCESSCODE_CANCELDELAYMESSAGE_REQUEST) )
	{
		nret = app_CancelDelayMessage( p_env , p_accepted_session ) ;
		if( nret == TCPMAIN_RETURN_CLOSE )
		{
			ERRORLOG( "app_CancelDelayMessage return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
		else
		{
			INFOLOG( "app_CancelDelayMessage return [%s]" , g_tcpdaemon_tcpmain_return_value_desc[nret] )
		}
	}
	else
	{
		ERRORLOG( "invalid process_code[%.4s][0x%X 0x%X 0x%X 0x%X]" , process_code.str , process_code.str[0],process_code.str[1],process_code.str[2],process_code.str[3] )
		return TCPMAIN_RETURN_CLOSE;
	}
	
	return nret;
}

/*
CREATE STORE :
     CSQ
APP ----> BROKER
    <----
     CSP

REQUEST MESSAGE PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|store_name_len(1bytes)|store_name(store_name_len bytes)|
|'0'|" CSQ"|0x07|"MYSTORE"|

RESPONSE MESSAGE PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|result(2bytes)|
|'0'|" CSP"|0x0000|
*/

int app_CreateStore( struct TcpdaemonServerEnvironment *p_env , struct MQAcceptedSession *p_accepted_session )
{
	struct MQBroker	*broker = p_accepted_session->broker ;
	
	uint8_t			store_name_len ;
	char			*store_name = NULL ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	
	struct MQStore		*p_store = NULL ;
	struct MQFileMapping	*msg_file = NULL ;
	struct MQFileMapping	*delay_msg_file = NULL ;
	
	int			nret = 0 ;
	
	MEMUNPACK_STORE_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len )
	MEMUNPACK_STORE_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len , store_name )
	
	INFOLOG( "RECV CREATE STORE REQUEST - store_name[%.*s]" , store_name_len,store_name )
	
	ResetAppendBufferFillPtr( p_accepted_session->comm_data ) ;
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return TCPMAIN_RETURN_CLOSE , comm_version , p_accepted_session->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_CREATESTORE_RESPONSE ;
	MEMPACK_PROCESS_CODE( return TCPMAIN_RETURN_CLOSE , process_code , p_accepted_session->comm_data )
	
	do
	{
		p_store = (struct MQStore *)malloc( sizeof(struct MQStore) ) ;
		if( p_store == NULL )
		{
			ERRORLOG( "malloc MQStore failed , errno[%d]" , errno )
			return TCPMAIN_RETURN_CLOSE;
		}
		memset( p_store , 0x00 , sizeof(struct MQStore) );
		
		strncpy( p_store->store_name , store_name , store_name_len );
		p_store->store_name_len = store_name_len ;
		
		memset( p_store->store_path , 0x00 , sizeof(p_store->store_path) );
		snprintf( p_store->store_path , sizeof(p_store->store_path)-1 , "%s/%.*s" , broker->data_path , p_store->store_name_len,p_store->store_name );
		if( mkdir( p_store->store_path , S_IRWXU ) == -1 )
		{
			ERRORLOG( "make store_path[%s] failed , errno[%d]" , p_store->store_path , errno )
			free( p_store );
			MEMPACK_RESULT( break , MQSDK_ERROR_MAKE_STORE_PATH , p_accepted_session->comm_data )
			break;
		}
		else
		{
			INFOLOG( "make store_path[%s] ok" , p_store->store_path )
		}
		
		p_store->max_msg_offset = 0 ;
		INIT_LIST_HEAD( & (p_store->msg_file_list) );
		INIT_LIST_HEAD( & (p_store->delay_msg_file_list) );
		
		DEBUGLOG( "add store[%.*s] to store_name_tree ..." , p_store->store_name_len,p_store->store_name )
		nret = LinkStoreNameTreeNode( broker , p_store ) ;
		if( nret )
		{
			FATALLOG( "LinkStoreNameTreeNode failed , store_name[%s]" , p_store->store_name )
			free( p_store );
			MEMPACK_RESULT( break , MQSDK_ERROR_INTERNAL , p_accepted_session->comm_data )
			break;
		}
		else
		{
			DEBUGLOG( "LinkStoreNameTreeNode ok , store_name[%s]" , p_store->store_name )
		}
		
		nret = CreateMessageDataFile( p_accepted_session->comm_data , broker , p_store , 0 , & msg_file , NULL ) ;
		if( nret )
		{
			ERRORLOG( "CreateMessageDataFile failed" )
			rmdir( p_store->store_path );
			free( p_store );
			MEMPACK_RESULT( break , MQSDK_ERROR_CREATE_MSG_DATA_FILE , p_accepted_session->comm_data )
			break;
		}
		else
		{
			INFOLOG( "CreateMessageDataFile ok" )
		}
		
		nret = CreateDelayMessageDataFile( p_accepted_session->comm_data , broker , p_store , 0 , & delay_msg_file , NULL ) ;
		if( nret )
		{
			ERRORLOG( "CreateDelayMessageDataFile failed" )
			rmdir( p_store->store_path );
			free( p_store );
			MEMPACK_RESULT( break , MQSDK_ERROR_CREATE_DELAY_MSG_FILE , p_accepted_session->comm_data )
			break;
		}
		else
		{
			INFOLOG( "CreateDelayMessageDataFile ok" )
		}
		
		MEMPACK_RESULT( break , MQSDK_SUCCESS , p_accepted_session->comm_data )
	}
	while(0);
	
	DEBUGHEXLOG( GetAppendBufferBase(p_accepted_session->comm_data) , GetAppendBufferLength(p_accepted_session->comm_data) , "pack ok , [%zu]bytes" , GetAppendBufferLength(p_accepted_session->comm_data) )
	
	return TCPMAIN_RETURN_WAITINGFOR_SENDING;
}

/*
DESTROY STORE :
     DSQ
APP ----> BROKER
    <----
     DSP

REQUEST MESSAGE PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|store_name_len(1bytes)|store_name(store_name_len bytes)|
|'0'|" DSQ"|0x07|"MYSTORE"|

RESPONSE MESSAGE PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|result(2bytes)|
|'0'|" DSP"|0x0000|
*/

int app_DestroyStore( struct TcpdaemonServerEnvironment *p_env , struct MQAcceptedSession *p_accepted_session )
{
	struct MQBroker	*broker = p_accepted_session->broker ;
	
	uint8_t			store_name_len ;
	char			*store_name = NULL ;
	char			store_path[ PATH_MAX ] ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	
	struct MQStore		store ;
	struct MQStore		*p_store = NULL ;
	struct MQFileMapping	*msg_file = NULL ;
	struct MQFileMapping	*next_msg_file = NULL ;
	
	int			nret = 0 ;
	
	MEMUNPACK_STORE_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len )
	MEMUNPACK_STORE_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len , store_name )
	
	INFOLOG( "RECV DESTROY STORE REQUEST - store_name[%.*s]" , store_name_len,store_name )
	
	ResetAppendBufferFillPtr( p_accepted_session->comm_data ) ;
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return TCPMAIN_RETURN_CLOSE , comm_version , p_accepted_session->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_DESTROYSTORE_RESPONSE ;
	MEMPACK_PROCESS_CODE( return TCPMAIN_RETURN_CLOSE , process_code , p_accepted_session->comm_data )
	
	do
	{
		memset( & store , 0x00 , sizeof(struct MQStore) );
		strncpy( store.store_name , store_name , store_name_len );
		store.store_name_len = store_name_len ;
		p_store = QueryStoreNameWithLenTreeNode( broker , & store ) ;
		if( p_store == NULL )
		{
			ERRORLOG( "store[%s] not exist" , store.store_name )
			MEMPACK_RESULT( break , MQSDK_ERROR_STORE_NOT_EXIST , p_accepted_session->comm_data )
			break;
		}
		else
		{
			DEBUGLOG( "store[%s] exist" , store.store_name )
		}
		
		list_for_each_entry_safe( msg_file , next_msg_file , & (p_store->msg_file_list) , struct MQFileMapping , file_list_node )
		{
			DEBUGLOG( "unlink from msg_file_list" )
			list_del( & (msg_file->file_list_node) );
			
			DEBUGLOG( "munmap[0x%X]" , msg_file->file_mmap )
			munmap( msg_file->file_mmap , msg_file->file_size );
			
			DEBUGLOG( "unlink[%s]" , msg_file->path_file_name )
			unlink( msg_file->path_file_name );
		}
		
		UnlinkStoreNameTreeNode( broker , p_store );
		free( p_store );
		
		memset( store_path , 0x00 , sizeof(store_path) );
		snprintf( store_path , sizeof(store_path)-1 , "%s/%.*s" , broker->data_path , store_name_len,store_name );
		if( rmdir( store_path ) == -1 )
		{
			ERRORLOG( "remove store_path[%s] failed , errno[%d]" , store_path , errno )
			MEMPACK_RESULT( break , MQSDK_ERROR_REMOVE_STORE_PATH , p_accepted_session->comm_data )
			break;
		}
		else
		{
			INFOLOG( "remove store_path[%s] ok" , store_path )
		}
		
		MEMPACK_RESULT( break , MQSDK_SUCCESS , p_accepted_session->comm_data )
	}
	while(0);
	
	DEBUGHEXLOG( GetAppendBufferBase(p_accepted_session->comm_data) , GetAppendBufferLength(p_accepted_session->comm_data) , "pack ok , [%zu]bytes" , GetAppendBufferLength(p_accepted_session->comm_data) )
	
	return TCPMAIN_RETURN_WAITINGFOR_SENDING;
}

/*
CREATE TOPIC :
     CTQ
APP ----> BROKER
    <----
     CTP

REQUEST MESSAGE PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|store_name_len(1bytes)|store_name(store_name_len bytes)|topic_name_len(1bytes)|topic(topic_name_len bytes)|
|'0'|" CTQ"|0x07|"MYSTORE"|0x07|"MYTOPIC"|

RESPONSE MESSAGE PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|result(2bytes)|
|'0'|" CTP"|0x0000|
*/

int app_CreateTopic( struct TcpdaemonServerEnvironment *p_env , struct MQAcceptedSession *p_accepted_session )
{
	struct MQBroker	*broker = p_accepted_session->broker ;
	
	uint8_t			store_name_len ;
	char			*store_name = NULL ;
	uint8_t			topic_name_len ;
	char			*topic_name = NULL ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	
	struct MQStore		store ;
	struct MQStore		*p_store = NULL ;
	struct MQTopic		topic ;
	struct MQTopic		*p_topic = NULL ;
	
	int			nret = 0 ;
	
	MEMUNPACK_STORE_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len )
	MEMUNPACK_STORE_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len , store_name )
	
	MEMUNPACK_TOPIC_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , topic_name_len )
	MEMUNPACK_TOPIC_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , topic_name_len , topic_name )
	
	INFOLOG( "RECV CREATE TOPIC REQUEST - store_name[%.*s] topic_name[%.*s]" , store_name_len,store_name , topic_name_len,topic_name )
	
	ResetAppendBufferFillPtr( p_accepted_session->comm_data ) ;
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return TCPMAIN_RETURN_CLOSE , comm_version , p_accepted_session->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_CREATETOPIC_RESPONSE ;
	MEMPACK_PROCESS_CODE( return TCPMAIN_RETURN_CLOSE , process_code , p_accepted_session->comm_data )
	
	do
	{
		memset( & store , 0x00 , sizeof(struct MQStore) );
		strncpy( store.store_name , store_name , store_name_len );
		store.store_name_len = store_name_len ;
		p_store = QueryStoreNameWithLenTreeNode( broker , & store ) ;
		if( p_store == NULL )
		{
			ERRORLOG( "store[%s] not exist" , store.store_name )
			MEMPACK_RESULT( break , MQSDK_ERROR_STORE_NOT_EXIST , p_accepted_session->comm_data )
			break;
		}
		else
		{
			DEBUGLOG( "store[%s] exist" , store.store_name )
		}
		
		memset( & topic , 0x00 , sizeof(struct MQTopic) );
		strncpy( topic.topic_name , topic_name , topic_name_len );
		topic.topic_name_len = topic_name_len ;
		p_topic = QueryTopicNameWithLenTreeNode( p_store , & topic ) ;
		if( p_topic != NULL )
		{
			ERRORLOG( "topic[%s] exist" , topic.topic_name )
			MEMPACK_RESULT( break , MQSDK_ERROR_TOPIC_EXIST_IN_STORE , p_accepted_session->comm_data )
			break;
		}
		else
		{
			DEBUGLOG( "topic[%s] not exist" , topic.topic_name )
		}
		
		p_topic = (struct MQTopic *)malloc( sizeof(struct MQTopic) ) ;
		if( p_topic == NULL )
		{
			ERRORLOG( "malloc MQTopic failed , errno[%d]" , errno )
			return TCPMAIN_RETURN_CLOSE;
		}
		memset( p_topic , 0x00 , sizeof(struct MQTopic) );
		
		strncpy( p_topic->topic_name , topic_name , topic_name_len );
		p_topic->topic_name_len = topic_name_len ;
		
		memset( p_topic->topic_path , 0x00 , sizeof(p_topic->topic_path) );
		snprintf( p_topic->topic_path , sizeof(p_topic->topic_path)-1 , "%s/%.*s" , p_store->store_path , p_topic->topic_name_len,p_topic->topic_name );
		if( mkdir( p_topic->topic_path , S_IRWXU ) == -1 )
		{
			ERRORLOG( "make topic_path[%s] failed , errno[%d]" , p_topic->topic_path , errno )
			free( p_topic );
			MEMPACK_RESULT( break , MQSDK_ERROR_MAKE_TOPIC_PATH , p_accepted_session->comm_data )
			break;
		}
		else
		{
			INFOLOG( "make topic_path[%s] ok" , p_topic->topic_path )
		}
		
		p_topic->max_topic_offset = 0 ;
		INIT_LIST_HEAD( & (p_topic->topic_file_list) );
		
		DEBUGLOG( "add topic[%.*s] to topic_name_tree" , p_topic->topic_name_len,p_topic->topic_name )
		nret = LinkTopicNameTreeNode( p_store , p_topic ) ;
		if( nret )
		{
			FATALLOG( "LinkTopicNameTreeNode failed , store_name[%s]" , p_topic->topic_name )
			free( p_topic );
			MEMPACK_RESULT( break , MQSDK_ERROR_INTERNAL , p_accepted_session->comm_data )
			break;
		}
		
		nret = CreateTopicDataFile( p_accepted_session->comm_data , broker , p_store , p_topic , 0 , NULL , NULL ) ;
		if( nret )
		{
			ERRORLOG( "CreateTopicDataFile failed" )
			rmdir( p_topic->topic_path );
			free( p_topic );
			MEMPACK_RESULT( break , MQSDK_ERROR_LOAD_TOPIC_FILE , p_accepted_session->comm_data )
			break;
		}
		else
		{
			INFOLOG( "CreateTopicDataFile ok" )
		}
		
		MEMPACK_RESULT( break , MQSDK_SUCCESS , p_accepted_session->comm_data )
	}
	while(0);
	
	DEBUGHEXLOG( GetAppendBufferBase(p_accepted_session->comm_data) , GetAppendBufferLength(p_accepted_session->comm_data) , "pack ok , [%zu]bytes" , GetAppendBufferLength(p_accepted_session->comm_data) )
	
	return TCPMAIN_RETURN_WAITINGFOR_SENDING;
}

/*
DESTROY TOPIC :
     DTQ
APP ----> BROKER
    <----
     DTP

REQUEST MESSAGE PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|store_name_len(1bytes)|store_name(store_name_len bytes)|topic_name_len(1bytes)|topic(topic_name_len bytes)|
|'0'|" DTQ"|0x07|"MYSTORE"|0x07|"MYTOPIC"|

RESPONSE MESSAGE PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|result(2bytes)|
|'0'|" DTP"|0x0000|
*/

int app_DestroyTopic( struct TcpdaemonServerEnvironment *p_env , struct MQAcceptedSession *p_accepted_session )
{
	struct MQBroker	*broker = p_accepted_session->broker ;
	
	uint8_t			store_name_len ;
	char			*store_name = NULL ;
	uint8_t			topic_name_len ;
	char			*topic_name = NULL ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	
	struct MQStore		store ;
	struct MQStore		*p_store = NULL ;
	struct MQTopic		topic ;
	struct MQTopic		*p_topic = NULL ;
	struct MQFileMapping	*topic_file = NULL ;
	struct MQFileMapping	*next_topic_file = NULL ;
	
	int			nret = 0 ;
	
	MEMUNPACK_STORE_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len )
	MEMUNPACK_STORE_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len , store_name )
	
	MEMUNPACK_TOPIC_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , topic_name_len )
	MEMUNPACK_TOPIC_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , topic_name_len , topic_name )
	
	INFOLOG( "RECV DESTROY TOPIC REQUEST - store_name[%.*s] topic_name[%.*s]" , store_name_len,store_name , topic_name_len,topic_name )
	
	ResetAppendBufferFillPtr( p_accepted_session->comm_data ) ;
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return TCPMAIN_RETURN_CLOSE , comm_version , p_accepted_session->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_DESTROYTOPIC_RESPONSE ;
	MEMPACK_PROCESS_CODE( return TCPMAIN_RETURN_CLOSE , process_code , p_accepted_session->comm_data )
	
	do
	{
		memset( & store , 0x00 , sizeof(struct MQStore) );
		strncpy( store.store_name , store_name , store_name_len );
		store.store_name_len = store_name_len ;
		p_store = QueryStoreNameWithLenTreeNode( broker , & store ) ;
		if( p_store == NULL )
		{
			ERRORLOG( "store[%s] not exist" , store.store_name )
			MEMPACK_RESULT( break , MQSDK_ERROR_STORE_NOT_EXIST , p_accepted_session->comm_data )
			break;
		}
		else
		{
			DEBUGLOG( "store[%s] exist" , store.store_name )
		}
		
		memset( & topic , 0x00 , sizeof(struct MQTopic) );
		strncpy( topic.topic_name , topic_name , topic_name_len );
		topic.topic_name_len = topic_name_len ;
		p_topic = QueryTopicNameWithLenTreeNode( p_store , & topic ) ;
		if( p_topic == NULL )
		{
			ERRORLOG( "topic[%s] not exist" , topic.topic_name )
			MEMPACK_RESULT( break , MQSDK_ERROR_TOPIC_NOT_EXIST_IN_STORE , p_accepted_session->comm_data )
			break;
		}
		else
		{
			DEBUGLOG( "topic[%s] exist" , topic.topic_name )
		}
		
		list_for_each_entry_safe( topic_file , next_topic_file , & (p_topic->topic_file_list) , struct MQFileMapping , file_list_node )
		{
			DEBUGLOG( "unlink from topic_file_list" )
			list_del( & (topic_file->file_list_node) );
			
			DEBUGLOG( "munmap[0x%X]" , topic_file->file_mmap )
			munmap( topic_file->file_mmap , topic_file->file_size );
			
			DEBUGLOG( "unlink[%s]" , topic_file->path_file_name )
			unlink( topic_file->path_file_name );
		}
		
		if( rmdir( p_topic->topic_path ) == -1 )
		{
			ERRORLOG( "remove topic_path[%s] failed , errno[%d]" , p_topic->topic_path , errno )
			MEMPACK_RESULT( break , MQSDK_ERROR_REMOVE_TOPIC_PATH , p_accepted_session->comm_data )
			break;
		}
		else
		{
			INFOLOG( "remove topic_path[%s] ok" , p_topic->topic_path )
		}
		
		UnlinkTopicNameTreeNode( p_store , p_topic );
		free( p_topic );
		
		MEMPACK_RESULT( break , MQSDK_SUCCESS , p_accepted_session->comm_data )
	}
	while(0);
	
	DEBUGHEXLOG( GetAppendBufferBase(p_accepted_session->comm_data) , GetAppendBufferLength(p_accepted_session->comm_data) , "pack ok , [%zu]bytes" , GetAppendBufferLength(p_accepted_session->comm_data) )
	
	return TCPMAIN_RETURN_WAITINGFOR_SENDING;
}

int FetchSubscribeGroupMessageToAcceptedSession( struct TcpdaemonServerEnvironment *p_env , struct MQSubscribeGroup *p_subscribe_group , struct MQAcceptedSession *p_accepted_session )
{
	struct MQFileMapping	*next_topic_file = NULL ;
	uint64_t		topic_offset ;
	uint64_t		msg_offset ;
	char			*msg_offset_ptr = NULL ;
	ssize_t			read_len ;
	
	uint64_t		msg_id ;
	uint8_t			topic_name_len ;
	char			*topic_name ;
	uint16_t		tags_len ;
	char			*tags = NULL ;
	uint32_t		msg_data_len ;
	char			*msg_data = NULL ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	
	int			nret = 0 ;
	
	if( p_accepted_session->long_pull_as_server_status != MQ_LONG_PULL_AS_SERVER_STATUS_SILENT )
	{
		DEBUGLOG( "long_pull_as_server_status[%d] != MQ_LONG_PULL_AS_SERVER_STATUS_SILENT[%d]" , p_accepted_session->long_pull_as_server_status , MQ_LONG_PULL_AS_SERVER_STATUS_SILENT )
		return MQ_BROKER_RETURN_SESSION_IS_BUSY;
	}
	
	for( ; ; )
	{
		DEBUGLOG( "MemRead_TopicDataUnit ... p_subscribe_group[%p]->topic_fill_ptr[%p]" , p_subscribe_group , p_subscribe_group->topic_fill_ptr )
		read_len = MemRead_TopicDataUnit( p_subscribe_group->topic_fill_ptr , & topic_offset , & msg_offset ) ;
		if( read_len == 0 )
		{
			DEBUGLOG( "topic file no append" )
			return MQ_BROKER_RETURN_FETCH_NOTHING;
		}
		else if( read_len == UINT64_MAX )
		{
			next_topic_file = list_next_entry( p_subscribe_group->topic_file , struct MQFileMapping , file_list_node ) ;
			if( & (next_topic_file->file_list_node) == & (p_subscribe_group->p_topic->topic_file_list) )
			{
				FATALLOG( "travel topic_file over" )
				return MQ_BROKER_RETURN_FATAL;
			}
			
			p_subscribe_group->topic_file = next_topic_file ;
			p_subscribe_group->topic_fill_ptr = p_subscribe_group->topic_file->file_mmap + MQ_TOPIC_FILE_HEAD_SIZE ;
		}
		else
		{
			DEBUGLOG( "MemRead_TopicDataUnit ok , p_subscribe_group[%p]->topic_fill_ptr[%p] msg_offset[%"PRIu64"]" , p_subscribe_group , p_subscribe_group->topic_fill_ptr , msg_offset )
			p_accepted_session->p_subscribe_group->topic_fill_ptr += read_len ;
			p_accepted_session->p_subscribe_group->topic_fill_step = read_len ;
			break;
		}
	}
	
	for( ; ; )
	{
		if( p_subscribe_group->msg_file->begin_offset <= msg_offset && msg_offset <= p_subscribe_group->msg_file->begin_offset+p_subscribe_group->msg_file->file_size-1 )
		{
			break;
		}
		
		p_subscribe_group->msg_file = list_next_entry( p_subscribe_group->msg_file , struct MQFileMapping , file_list_node ) ;
		if( & (p_subscribe_group->msg_file->file_list_node) == & (p_subscribe_group->p_store->msg_file_list) )
		{
			FATALLOG( "travel msg_file over" )
			return MQ_BROKER_RETURN_FATAL;
		}
	}
	
	msg_offset_ptr = p_subscribe_group->msg_file->file_mmap + (msg_offset-p_subscribe_group->msg_file->begin_offset) ;
	read_len = MemRead_MessageDataUnit( msg_offset_ptr , & msg_id , & topic_name_len , & topic_name , & tags_len , & tags , & msg_data_len , & msg_data ) ;
	if( read_len == 0 )
	{
		FATALLOG( "msg file no append" )
		return MQ_BROKER_RETURN_FETCH_NOTHING;
	}
	
	ResetAppendBufferFillPtr( p_accepted_session->comm_data ) ;
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return TCPMAIN_RETURN_CLOSE , comm_version , p_accepted_session->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_CONSUMEMESSAGE_REQUEST ;
	MEMPACK_PROCESS_CODE( return TCPMAIN_RETURN_CLOSE , process_code , p_accepted_session->comm_data )
	
	MEMPACK_TOPIC_NAME_LEN( return TCPMAIN_RETURN_CLOSE , topic_name_len , p_accepted_session->comm_data );
	MEMPACK_TOPIC_NAME( return TCPMAIN_RETURN_CLOSE , topic_name , topic_name_len , p_accepted_session->comm_data );
	
	MEMPACK_TAGS_LEN( return TCPMAIN_RETURN_CLOSE , tags_len , p_accepted_session->comm_data );
	MEMPACK_TAGS( return TCPMAIN_RETURN_CLOSE , tags , tags_len , p_accepted_session->comm_data );
	
	MEMPACK_MSG_DATA_LEN( return TCPMAIN_RETURN_CLOSE , msg_data_len , p_accepted_session->comm_data );
	MEMPACK_MSG_DATA( return TCPMAIN_RETURN_CLOSE , msg_data , msg_data_len , p_accepted_session->comm_data );
	
	MEMPACK_TOPIC_OFFSET( return TCPMAIN_RETURN_CLOSE , topic_offset , p_accepted_session->comm_data );
	
	p_accepted_session->comm_status = TCPMAIN_RETURN_WAITINGFOR_SENDING ;
	p_accepted_session->long_pull_as_server_status = MQ_LONG_PULL_AS_SERVER_STATUS_FIRST_SEND_BEFORE_SILENT ;
	
	TDModifySessionOutputEvent( p_env , p_accepted_session->tcpdaemon_session );
	
	return MQ_BROKER_RETURN_FETCH_ONE;
}

int FetchSubscribeGroupMessage( struct TcpdaemonServerEnvironment *p_env , struct MQSubscribeGroup *p_subscribe_group )
{
	struct MQAcceptedSession	*p_subscribe_session = NULL ;
	struct MQAcceptedSession	*p_next_subscribe_session = NULL ;
	struct MQAcceptedSession	*p_last_subscribe_session = NULL ;
	
	int				nret = 0 ;
	
	p_subscribe_session = list_first_entry( & (p_subscribe_group->subscribe_session_list) , struct MQAcceptedSession , subscribe_group_subscribe_session_list_node ) ;
	p_last_subscribe_session = list_last_entry( & (p_subscribe_group->subscribe_session_list) , struct MQAcceptedSession , subscribe_group_subscribe_session_list_node ) ;
	DEBUGLOG( "p_subscribe_group[%p][%s] p_first_subscribe_session[%p] p_last_subscribe_session[%p]" , p_subscribe_group,p_subscribe_group->subscribe_group_name , p_subscribe_session , p_last_subscribe_session )
	
	for( ; ; )
	{
		p_next_subscribe_session = list_next_entry( p_subscribe_session , struct MQAcceptedSession , subscribe_group_subscribe_session_list_node ) ;
		
		DEBUGLOG( "FetchSubscribeGroupMessageToAcceptedSession p_subscribe_session[%p] ..." , p_subscribe_session )
		nret = FetchSubscribeGroupMessageToAcceptedSession( p_env , p_subscribe_session->p_subscribe_group , p_subscribe_session ) ;
		if( nret < 0 )
		{
			ERRORLOG( "FetchSubscribeGroupMessageToAcceptedSession p_subscribe_session[%p] failed[%d]" , p_subscribe_session , nret )
			return nret;
		}
		else if( nret == MQ_BROKER_RETURN_FETCH_ONE )
		{
			INFOLOG( "FetchSubscribeGroupMessageToAcceptedSession p_subscribe_session[%p] ok , fetch one" , p_subscribe_session )
			if( p_subscribe_session != p_last_subscribe_session )
			{
				list_del( & (p_subscribe_session->subscribe_group_subscribe_session_list_node) );
				list_add_tail( & (p_subscribe_session->subscribe_group_subscribe_session_list_node) , & (p_subscribe_group->subscribe_session_list) );
			}
			break;
		}
		else if( nret == MQ_BROKER_RETURN_FETCH_NOTHING )
		{
			INFOLOG( "FetchSubscribeGroupMessageToAcceptedSession p_subscribe_session[%p] ok , nothing" , p_subscribe_session )
		}
		else
		{
			INFOLOG( "FetchSubscribeGroupMessageToAcceptedSession p_subscribe_session[%p] ok , maybe session is busy" , p_subscribe_session )
		}
		
		if( p_subscribe_session == p_last_subscribe_session )
			break;
		p_subscribe_session = p_next_subscribe_session ;
		DEBUGLOG( "p_subscribe_group[%p][%s] p_next_subscribe_session[%p] p_last_subscribe_session[%p]" , p_subscribe_group,p_subscribe_group->subscribe_group_name , p_subscribe_session , p_last_subscribe_session )
	}
	
	return 0;
}

int FetchTopicMessage( struct TcpdaemonServerEnvironment *p_env , struct MQTopic *topic )
{
	struct MQSubscribeGroup		*p_subscribe_group = NULL ;
	
	int				nret = 0 ;
	
	p_subscribe_group = NULL ;
	for( ; ; )
	{
		p_subscribe_group = TravelSubscribeGroupNameTreeNode( topic , p_subscribe_group ) ;
		if( p_subscribe_group == NULL )
			break;
		
		nret = FetchSubscribeGroupMessage( p_env , p_subscribe_group ) ;
		if( nret )
			return nret;
	}
	
	return 0;
}

static int app_AddMessage( struct TcpdaemonServerEnvironment *p_env , struct MQAcceptedSession *p_accepted_session , struct AppendBuffer *comm_data , uint8_t store_name_len , char *store_name , uint8_t topic_name_len , char *topic_name , uint16_t tags_len , char *tags , uint32_t msg_data_len , char *msg_data )
{
	struct MQBroker			*broker = (struct MQBroker *)TDGetTcpmainParameter(p_env) ;
	
	struct MQStore			store ;
	struct MQStore			*p_store = NULL ;
	struct MQTopic			topic ;
	struct MQTopic			*p_topic = NULL ;
	struct MQFileMapping		*last_msg_file = NULL ;
	size_t				total_msg_data_unit_size ;
	ssize_t				wrote_len ;
	uint64_t			msg_offset = 0 ;
	struct MQFileMapping		*last_topic_file = NULL ;
	size_t				total_topic_unit_size ;
	uint64_t			topic_offset ;
	
	struct MQFileMapping		*first_msg_file = NULL ;
	struct MQFileMapping		*first_topic_file = NULL ;
	struct MQAcceptedSession	*p_session = NULL ;
	struct MQAcceptedSession	*p_next_session = NULL ;
	
	int				nret = 0 ;
	
	INFOLOG( "app_AddMessage - store_name[%.*s] topic_name[%.*s] tags[%.*s] msg_data[%.*s]" , store_name_len,store_name , topic_name_len,topic_name , tags_len,tags , msg_data_len,msg_data )
	
	do
	{
		memset( & store , 0x00 , sizeof(struct MQStore) );
		strncpy( store.store_name , store_name , store_name_len );
		store.store_name_len = store_name_len ;
		p_store = QueryStoreNameWithLenTreeNode( broker , & store ) ;
		if( p_store == NULL )
		{
			ERRORLOG( "store[%s] not exist" , store.store_name )
			MEMPACK_RESULT( break , MQSDK_ERROR_STORE_NOT_EXIST , comm_data )
			break;
		}
		else
		{
			DEBUGLOG( "store[%s] exist" , store.store_name )
		}
		
		memset( & topic , 0x00 , sizeof(struct MQTopic) );
		strncpy( topic.topic_name , topic_name , topic_name_len );
		topic.topic_name_len = topic_name_len ;
		p_topic = QueryTopicNameWithLenTreeNode( p_store , & topic ) ;
		if( p_topic == NULL )
		{
			ERRORLOG( "topic[%s] not exist" , topic.topic_name )
			MEMPACK_RESULT( break , MQSDK_ERROR_STORE_NOT_EXIST , comm_data )
			break;
		}
		else
		{
			DEBUGLOG( "topic[%s] exist" , topic.topic_name )
		}
		
		last_msg_file = list_last_entry( & (p_store->msg_file_list) , struct MQFileMapping , file_list_node ) ;
		total_msg_data_unit_size = CalcMessageDataUnitSize( topic_name_len , tags_len , msg_data_len ) ;
		DEBUGLOG( "last_msg_file->fill_ptr[0x%X] + sizeof(uint64_t)[%zu] + total_msg_data_unit_size[%zu] = [0x%X] > last_msg_file->over_end_ptr[0x%X] ? [%d]" , last_msg_file->fill_ptr , sizeof(uint64_t) , total_msg_data_unit_size , last_msg_file->fill_ptr+sizeof(uint64_t)+total_msg_data_unit_size , last_msg_file->over_end_ptr , last_msg_file->fill_ptr+sizeof(uint64_t)+total_msg_data_unit_size>last_msg_file->over_end_ptr )
		if( last_msg_file->fill_ptr + total_msg_data_unit_size + sizeof(uint64_t) > last_msg_file->over_end_ptr )
		{
			uint64_t	new_msg_file_begin_offset = last_msg_file->begin_offset + last_msg_file->file_size ;
			
			*(uint64_t*)(last_msg_file->fill_ptr) = hton64(UINT64_MAX) ;
			
			p_store->max_msg_offset += last_msg_file->over_end_ptr - last_msg_file->fill_ptr ;
			DEBUGLOG( "set p_store->max_msg_offset[%zu]" , p_store->max_msg_offset )
			
			nret = CreateMessageDataFile( comm_data , broker , p_store , new_msg_file_begin_offset , & last_msg_file , NULL ) ;
			if( nret )
			{
				ERRORLOG( "CreateMessageDataFile failed , new_msg_file_begin_offset[%"PRIu64"]" , new_msg_file_begin_offset )
				MEMPACK_RESULT( break , MQSDK_ERROR_LOAD_MSG_DATA_FILE , comm_data )
				break;
			}
			else
			{
				INFOLOG( "CreateMessageDataFile ok , new_msg_file_begin_offset[%"PRIu64"]" , new_msg_file_begin_offset )
			}
		}
		
		msg_offset = last_msg_file->fill_ptr - last_msg_file->file_mmap + last_msg_file->begin_offset ;
		INFOLOG( "before writing msg_file , msg_offset[%"PRIu64"] p_store->max_msg_offset[%"PRIu64"] msg_offset[%"PRIu64"]" , msg_offset , p_store->max_msg_offset )
		wrote_len = MemWrite_MessageDataUnit( last_msg_file->fill_ptr , msg_offset , topic_name_len , topic_name , tags_len , tags , msg_data_len , msg_data ) ;
		DEBUGHEXLOG( last_msg_file->fill_ptr , wrote_len , "wrote store_name[%.*s] topic_name[%.*s] tags[%.*s] msg_data[%.*s] to msg_file[%s][%zu] in offset[%"PRIu64"] fill[%zu]bytes" , store_name_len,store_name , topic_name_len,topic_name , tags_len,tags , msg_data_len,msg_data , last_msg_file->path_file_name , last_msg_file->begin_offset , msg_offset , wrote_len )
		last_msg_file->fill_ptr += wrote_len ;
		p_store->max_msg_offset = wrote_len ;
		INFOLOG( "after writing msg_file ,  msg_offset[%"PRIu64"] p_store->max_msg_offset[%"PRIu64"] msg_offset[%"PRIu64"]" , msg_offset , p_store->max_msg_offset )
		
		if( p_accepted_session )
		{
			INFOLOG( "del and add accepted_session[%p] to last_msg_file[%s]->msg_file_subscribe_session_list" , p_accepted_session , last_msg_file->path_file_name )
			if( ! list_empty( & (p_accepted_session->msg_file_subscribe_session_list_node) ) )
				list_del( & (p_accepted_session->msg_file_subscribe_session_list_node) );
			list_add_tail( & (p_accepted_session->msg_file_subscribe_session_list_node) , & (last_msg_file->subscribe_session_list) );
		}
		
		last_topic_file = list_last_entry( & (p_topic->topic_file_list) , struct MQFileMapping , file_list_node ) ;
		total_topic_unit_size = CalcTopicDataUnitSize() ;
		DEBUGLOG( "last_topic_file->fill_ptr[0x%X] + sizeof(uint64_t)[%zu] + total_topic_unit_size[%zu] = [0x%X] > last_topic_file->over_end_ptr[0x%X] ? [%d]" , last_topic_file->fill_ptr , sizeof(uint64_t) , total_topic_unit_size , last_topic_file->fill_ptr+sizeof(uint64_t)+total_topic_unit_size , last_topic_file->over_end_ptr , last_topic_file->fill_ptr+sizeof(uint64_t)+total_topic_unit_size>last_topic_file->over_end_ptr )
		if( last_topic_file->fill_ptr + total_topic_unit_size + sizeof(uint64_t) > last_topic_file->over_end_ptr )
		{
			uint64_t	new_topic_file_begin_offset = last_topic_file->begin_offset + last_topic_file->file_size ;
			
			*(uint64_t*)(last_topic_file->fill_ptr) = hton64(UINT64_MAX) ;
			
			p_topic->max_topic_offset += last_topic_file->over_end_ptr - last_topic_file->fill_ptr ;
			DEBUGLOG( "set p_topic->max_topic_offset[%zu]" , p_topic->max_topic_offset )
			
			nret = CreateTopicDataFile( comm_data , broker , p_store , p_topic , new_topic_file_begin_offset , & last_topic_file , NULL ) ;
			if( nret )
			{
				ERRORLOG( "CreateTopicDataFile failed , new_topic_file_begin_offset[%"PRIu64"]" , new_topic_file_begin_offset )
				MEMPACK_RESULT( break , MQSDK_ERROR_LOAD_TOPIC_FILE , comm_data )
				break;
			}
			else
			{
				INFOLOG( "CreateTopicDataFile ok , new_topic_file_begin_offset[%"PRIu64"]" , new_topic_file_begin_offset )
			}
		}
		
		topic_offset = last_topic_file->fill_ptr - last_topic_file->file_mmap + last_topic_file->begin_offset ;
		INFOLOG( "before writing topic_file , topic_offset[%"PRIu64"] p_topic->max_topic_offset[%"PRIu64"]" , topic_offset , p_topic->max_topic_offset )
		wrote_len = MemWrite_TopicDataUnit( last_topic_file->fill_ptr , topic_offset , msg_offset ) ;
		last_topic_file->fill_ptr += wrote_len ;
		p_topic->max_topic_offset = wrote_len ;
		DEBUGHEXLOG( last_topic_file->fill_ptr , wrote_len , "wrote msg_offset[%"PRIu64"] to topic_file[%s][%zu] in offset[%"PRIu64"] fill[%zu]bytes" , msg_offset , last_topic_file->path_file_name , last_topic_file->begin_offset , topic_offset , wrote_len )
		INFOLOG( "after writing topic_file ,  topic_offset[%"PRIu64"] p_topic->max_topic_offset[%"PRIu64"]" , topic_offset , p_topic->max_topic_offset )
		
		if( p_accepted_session )
		{
			INFOLOG( "del and add accepted_session[%p] to last_topic_file[%s]->topic_file_subscribe_session_list" , p_accepted_session , last_topic_file->path_file_name )
			if( ! list_empty( & (p_accepted_session->topic_file_subscribe_session_list_node) ) )
				list_del( & (p_accepted_session->topic_file_subscribe_session_list_node) );
			list_add_tail( & (p_accepted_session->topic_file_subscribe_session_list_node) , & (last_topic_file->subscribe_session_list) );
		}
		
		nret = FetchTopicMessage( p_env , p_topic ) ;
		if( nret < 0 )
		{
			FATALLOG( "FetchTopicMessage p_topic[%s] failed[%d]" , p_topic->topic_name , nret )
			return TCPMAIN_RETURN_CLOSE;
		}
		else
		{
			INFOLOG( "FetchTopicMessage p_topic[%s] ok" , p_topic->topic_name )
		}
		
		MEMPACK_RESULT( break , MQSDK_SUCCESS , comm_data )
	}
	while(0);
	
	MEMPACK_MSG_OFFSET( return TCPMAIN_RETURN_CLOSE , msg_offset , p_accepted_session->comm_data )
	
	if( comm_data )
	{
		DEBUGHEXLOG( GetAppendBufferBase(comm_data) , GetAppendBufferLength(comm_data) , "pack ok , [%zu]bytes" , GetAppendBufferLength(comm_data) )
	}
	
	if( p_store )
	{
		DEBUGLOG( "p_store->msg_file_list_nodes_count[%zu] > broker->msg_file_list_nodes_soft_max[%zu] ?" , p_store->msg_file_list_nodes_count , broker->msg_file_list_nodes_soft_max )
		if( p_store->msg_file_list_nodes_count > broker->msg_file_list_nodes_soft_max )
		{
			first_msg_file = list_first_entry( & (p_store->msg_file_list) , struct MQFileMapping , file_list_node ) ;
			if( first_msg_file )
			{
				DEBUGLOG( "list_first_entry msg_file->path_file_name[%s]" , first_msg_file->path_file_name )
				if( list_empty( & (first_msg_file->subscribe_session_list) ) )
				{
					DEBUGLOG( "CleanMessageDataFile" )
					CleanMessageDataFile( broker , p_store , first_msg_file );
				}
			}
		}
		
		DEBUGLOG( "p_store->msg_file_list_nodes_count[%zu] > broker->msg_file_list_nodes_hard_max[%zu] ?" , p_store->msg_file_list_nodes_count , broker->msg_file_list_nodes_hard_max )
		if( p_store->msg_file_list_nodes_count > broker->msg_file_list_nodes_hard_max )
		{
			first_msg_file = list_first_entry( & (p_store->msg_file_list) , struct MQFileMapping , file_list_node ) ;
			if( first_msg_file )
			{
				DEBUGLOG( "list_first_entry msg_file->path_file_name[%s]" , first_msg_file->path_file_name )
				list_for_each_entry_safe( p_session , p_next_session , & (first_msg_file->subscribe_session_list) , struct MQAcceptedSession , msg_file_subscribe_session_list_node )
				{
					DEBUGLOG( "CloseAcceptedSession psession[%p]" , p_session )
					CloseAcceptedSession( p_session );
				}
				
				DEBUGLOG( "CleanMessageDataFile" )
				CleanMessageDataFile( broker , p_store , first_msg_file );
			}
		}
	}
	
	if( p_topic )
	{
		DEBUGLOG( "p_topic->topic_file_list_nodes_count[%zu] > broker->topic_file_list_nodes_soft_max[%zu] ?" , p_topic->topic_file_list_nodes_count , broker->topic_file_list_nodes_soft_max )
		if( p_topic->topic_file_list_nodes_count > broker->topic_file_list_nodes_soft_max )
		{
			first_topic_file = list_first_entry( & (p_topic->topic_file_list) , struct MQFileMapping , file_list_node ) ;
			if( first_topic_file )
			{
				DEBUGLOG( "list_first_entry topic_file->path_file_name[%s]" , first_topic_file->path_file_name )
				if( list_empty( & (first_topic_file->subscribe_session_list) ) )
				{
					DEBUGLOG( "CleanTopicDataFile" )
					CleanTopicDataFile( broker , p_store , p_topic , first_topic_file );
				}
			}
		}
		
		DEBUGLOG( "p_topic->topic_file_list_nodes_count[%zu] > broker->topic_file_list_nodes_hard_max[%zu] ?" , p_topic->topic_file_list_nodes_count , broker->topic_file_list_nodes_soft_max )
		if( p_topic->topic_file_list_nodes_count > broker->topic_file_list_nodes_hard_max )
		{
			first_topic_file = list_first_entry( & (p_topic->topic_file_list) , struct MQFileMapping , file_list_node ) ;
			if( first_topic_file )
			{
				DEBUGLOG( "list_first_entry topic_file->path_file_name[%s]" , first_topic_file->path_file_name )
				list_for_each_entry_safe( p_session , p_next_session , & (first_topic_file->subscribe_session_list) , struct MQAcceptedSession , topic_file_subscribe_session_list_node )
				{
					DEBUGLOG( "CloseAcceptedSession psession[%p]" , p_session )
					TDCloseSession( p_session->p_tcpdaemon_env , p_session->tcpdaemon_session );
					CloseAcceptedSession( p_session );
				}
				
				DEBUGLOG( "CleanTopicDataFile" )
				CleanTopicDataFile( broker , p_store , p_topic , first_topic_file );
			}
		}
	}
	
	return TCPMAIN_RETURN_WAITINGFOR_SENDING;
}

/*
PRODUCE MESSAGE :
     PMQ
APP ----> BROKER
    <----
     PMP

|comm_version(1bytes)|process_code(4bytes)|store_name_len(1bytes)|store_name(store_name_len bytes)|topic_name_len(1bytes)|topic(topic_name_len bytes)|tags_len(2bytes)|tags(tags_len bytes)|msg_data_len(4bytes)|msg_data(msg_data_len bytes)|
|'0'|" PMQ"|0x07|"MYTOPIC"|0x06|"MYTAGS"|0x09|"MYMSGDATA"|

RESPONSE MESSAGE PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|result(2bytes)|delay_msg_offset(8bytes)|
|'0'|" PMP"|0x0000|12345678|
*/

int app_ProduceMessage( struct TcpdaemonServerEnvironment *p_env , struct MQAcceptedSession *p_accepted_session )
{
	uint8_t				store_name_len ;
	char				*store_name = NULL ;
	uint8_t				topic_name_len ;
	char				*topic_name = NULL ;
	uint16_t			tags_len ;
	char				*tags = NULL ;
	uint32_t			msg_data_len ;
	char				*msg_data = NULL ;
	
	uint8_t				comm_version ;
	union MQProcessCode		process_code ;
	
	int				nret = 0 ;
	
	MEMUNPACK_STORE_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len )
	MEMUNPACK_STORE_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len , store_name )
	
	MEMUNPACK_TOPIC_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , topic_name_len )
	MEMUNPACK_TOPIC_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , topic_name_len , topic_name )
	
	MEMUNPACK_TAGS_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , tags_len )
	MEMUNPACK_TAGS( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , tags_len , tags )
	
	MEMUNPACK_MSG_DATA_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , msg_data_len )
	MEMUNPACK_MSG_DATA( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , msg_data_len , msg_data )
	
	INFOLOG( "RECV PRODUCE MESSAGE REQUEST - store_name[%.*s] topic_name[%.*s] tags[%.*s] msg_data[%.*s]" , store_name_len,store_name , topic_name_len,topic_name , tags_len,tags , msg_data_len,msg_data )
	
	ResetAppendBufferFillPtr( p_accepted_session->comm_data ) ;
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return TCPMAIN_RETURN_CLOSE , comm_version , p_accepted_session->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_PRODUCEMESSAGE_RESPONSE ;
	MEMPACK_PROCESS_CODE( return TCPMAIN_RETURN_CLOSE , process_code , p_accepted_session->comm_data )
	
	return app_AddMessage( p_env , p_accepted_session , p_accepted_session->comm_data , store_name_len , store_name , topic_name_len , topic_name , tags_len , tags , msg_data_len , msg_data );
}

int app_AddDelayMessage( struct TcpdaemonServerEnvironment *p_env , struct MQDelayMessage *delay_msg )
{
	struct MQBroker		*broker = (struct MQBroker *)TDGetTcpmainParameter(p_env) ;
	
	uint64_t		unpack_len ;
	
	uint64_t		delay_msg_id ;
	uint32_t		delay_seconds ;
	char			is_sent ;
	uint8_t			store_name_len ;
	char			*store_name = NULL ;
	uint8_t			topic_name_len ;
	char			*topic_name = NULL ;
	uint16_t		tags_len ;
	char			*tags = NULL ;
	uint32_t		msg_data_len ;
	char			*msg_data = NULL ;
	
	struct MQStore		store ;
	struct MQStore		*p_store = NULL ;
	
	struct MQFileMapping	*next_not_sent_delay_msg_file = NULL ;
	struct MQDelayMessage	next_delay_msg ;
	
	int			nret = 0 ;
	
	do
	{
		unpack_len = MemRead_DelayMessageDataUnit( delay_msg->delay_msg_file_unit_base , & delay_msg_id , & delay_seconds , & is_sent , & store_name_len , & store_name , & topic_name_len , & topic_name , & tags_len , & tags , & msg_data_len , & msg_data ) ;
		if( unpack_len == 0 || unpack_len == UINT64_MAX )
		{
			FATALLOG( "MemRead_DelayMessageDataUnit failed , return[%"PRIu64"]" , unpack_len )
			break;
		}
		if( is_sent != MQ_DELAY_MSG_WAIT_TO_SEND )
		{
			WARNLOG( "is_sent[%c] is not MQ_DELAY_MSG_WAIT_TO_SEND" , is_sent )
			break;
		}
		
		nret = app_AddMessage( p_env , NULL , NULL , store_name_len , store_name , topic_name_len , topic_name , tags_len , tags , msg_data_len , msg_data ) ;
		if( nret != TCPMAIN_RETURN_WAITINGFOR_SENDING )
		{
			ERRORLOG( "app_AddMessage failed[%d]" , nret )
			break;
		}
		else
		{
			DEBUGLOG( "app_AddMessage ok" )
		}
	}
	while(0);
	
	memset( & store , 0x00 , sizeof(struct MQStore) );
	strncpy( store.store_name , store_name , store_name_len );
	store.store_name_len = store_name_len ;
	p_store = QueryStoreNameWithLenTreeNode( broker , & store ) ;
	if( p_store == NULL )
	{
		ERRORLOG( "store[%s] not exist" , store.store_name )
		return -2;
	}
	else
	{
		DEBUGLOG( "store[%s] exist" , store.store_name )
	}
	
	memcpy( & next_delay_msg , delay_msg , sizeof(struct MQDelayMessage) );
	
	while(1)
	{
		unpack_len = MemRead_DelayMessageDataUnit( next_delay_msg.delay_msg_file_unit_base , & delay_msg_id , & delay_seconds , & is_sent , & store_name_len , & store_name , & topic_name_len , & topic_name , & tags_len , & tags , & msg_data_len , & msg_data ) ;
		if( unpack_len == 0 )
		{
			if( delay_msg_id == 0 )
			{
				DEBUGLOG( "MemRead_DelayMessageDataUnit return[0]" )
				break;
			}
			else if( delay_msg_id == UINT64_MAX )
			{
				INFOLOG( "MemRead_DelayMessageDataUnit return[UINT64_MAX]" )
				
				if( next_delay_msg.delay_msg_file == p_store->last_not_sent_delay_msg_file )
				{
					DEBUGLOG( "travel last delay msg file" )
					break;
				}
				
				next_not_sent_delay_msg_file = list_next_entry( next_delay_msg.delay_msg_file , struct MQFileMapping , file_list_node ) ;
				if( next_delay_msg.delay_msg_file == p_store->first_not_sent_delay_msg_file )
				{
					DEBUGLOG( "set p_store->first_not_sent_delay_msg_file[%p] to [%p]" , p_store->first_not_sent_delay_msg_file , next_not_sent_delay_msg_file )
					p_store->first_not_sent_delay_msg_file = next_not_sent_delay_msg_file ;
				}
				
				next_delay_msg.delay_msg_file = next_not_sent_delay_msg_file ;
				next_delay_msg.delay_msg_file_unit_base = next_not_sent_delay_msg_file->file_mmap + MQ_DELAY_MSG_FILE_HEAD_SIZE ;
			}
		}
		else
		{
			if( is_sent == MQ_DELAY_MSG_WAIT_TO_SEND )
			{
				DEBUGLOG( "MemRead_DelayMessageDataUnit return[%"PRIu64"] , is_sent[%c] exist in file" , unpack_len , is_sent )
				break;
			}
			else
			{
				DEBUGLOG( "MemRead_DelayMessageDataUnit return[%"PRIu64"]" , unpack_len )
				next_delay_msg.delay_msg_file_unit_base += unpack_len ;
			}
		}
	}
	
	DEBUGLOG( "p_store->first_not_sent_delay_msg_file[%p] ->last_not_sent_delay_msg_file[%p]" , p_store->first_not_sent_delay_msg_file , p_store->last_not_sent_delay_msg_file )
	
	DEBUGLOG( "MemSet_DelayMessageDataUnit_is_sent MQ_DELAY_MSG_IS_SENT" )
	MemSet_DelayMessageDataUnit_is_sent( delay_msg->delay_msg_file_unit_base , MQ_DELAY_MSG_IS_SENT );
	
	DEBUGLOG( "UnlinkDelayMessageOffsetTreeNode delay_msg[%p]" , delay_msg )
	UnlinkDelayMessageOffsetTreeNode( p_store , delay_msg );
	DEBUGLOG( "UnlinkDelayMessageSendTimestampTreeNode delay_msg[%p]" , delay_msg )
	UnlinkDelayMessageSendTimestampTreeNode( p_store , delay_msg );
	
	DEBUGLOG( "free delay_msg[%p]" , delay_msg )
	free( delay_msg );
	
	return 0;
}

/*
SUBSCRIBE MESSAGE :
     STQ
APP ----> BROKER
    <----
     STP

REQUEST MESSAGE PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|store_name_len(1bytes)|store_name(store_name_len bytes)|topic_name_len(1bytes)|topic(topic_name_len bytes)|subscribe_group_name_len(1bytes)|subscribe_group_name(subscribe_group_name_len bytes)|
|'0'|" SMQ"|0x07|"MYTOPIC"|0x09|"MYSUBSGRP"

RESPONSE MESSAGE PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|result(2bytes)|
|'0'|" SMP"|0x0000|
*/

int app_SubscribeMessage( struct TcpdaemonServerEnvironment *p_env , struct MQAcceptedSession *p_accepted_session )
{
	struct MQBroker		*broker = p_accepted_session->broker ;
	
	uint8_t			store_name_len ;
	char			*store_name = NULL ;
	uint8_t			topic_name_len ;
	char			*topic_name = NULL ;
	uint8_t			subscribe_group_name_len ;
	char			*subscribe_group_name = NULL ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	
	struct MQStore		store ;
	struct MQStore		*p_store = NULL ;
	struct MQTopic		topic ;
	struct MQTopic		*p_topic = NULL ;
	struct MQSubscribeGroup	subscribe_group ;
	struct MQSubscribeGroup	*p_subscribe_group = NULL ;
	
	int			nret = 0 ;
	
	MEMUNPACK_STORE_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len )
	MEMUNPACK_STORE_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len , store_name )
	
	MEMUNPACK_TOPIC_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , topic_name_len )
	MEMUNPACK_TOPIC_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , topic_name_len , topic_name )
	
	MEMUNPACK_SUBSCRIBE_GROUP_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , subscribe_group_name_len )
	MEMUNPACK_SUBSCRIBE_GROUP_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , subscribe_group_name_len , subscribe_group_name )
	
	INFOLOG( "RECV SUBSCRIBE MESSAGE REQUEST - store_name[%.*s] topic_name[%.*s] subscribe_group_name[%.*s]" , store_name_len,store_name , topic_name_len,topic_name , subscribe_group_name_len,subscribe_group_name )
	
	ResetAppendBufferFillPtr( p_accepted_session->comm_data ) ;
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return TCPMAIN_RETURN_CLOSE , comm_version , p_accepted_session->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_SUBSCRIBEMESSAGE_RESPONSE ;
	MEMPACK_PROCESS_CODE( return TCPMAIN_RETURN_CLOSE , process_code , p_accepted_session->comm_data )
	
	do
	{
		memset( & store , 0x00 , sizeof(struct MQStore) );
		strncpy( store.store_name , store_name , store_name_len );
		store.store_name_len = store_name_len ;
		p_store = QueryStoreNameWithLenTreeNode( broker , & store ) ;
		if( p_store == NULL )
		{
			ERRORLOG( "store[%s] not exist" , store.store_name )
			MEMPACK_RESULT( break , MQSDK_ERROR_STORE_NOT_EXIST , p_accepted_session->comm_data )
			break;
		}
		else
		{
			DEBUGLOG( "store[%s] exist" , store.store_name )
		}
		
		memset( & topic , 0x00 , sizeof(struct MQTopic) );
		strncpy( topic.topic_name , topic_name , topic_name_len );
		topic.topic_name_len = topic_name_len ;
		p_topic = QueryTopicNameWithLenTreeNode( p_store , & topic ) ;
		if( p_topic == NULL )
		{
			ERRORLOG( "topic[%s] not exist" , topic.topic_name )
			MEMPACK_RESULT( break , MQSDK_ERROR_STORE_NOT_EXIST , p_accepted_session->comm_data )
			break;
		}
		else
		{
			DEBUGLOG( "topic[%s] exist" , topic.topic_name )
		}
		
		memset( & subscribe_group , 0x00 , sizeof(struct MQSubscribeGroup) );
		strncpy( subscribe_group.subscribe_group_name , subscribe_group_name , subscribe_group_name_len );
		subscribe_group.subscribe_group_name_len = subscribe_group_name_len ;
		p_subscribe_group = QuerySubscribeGroupNameWithLenTreeNode( p_topic , & subscribe_group ) ;
		if( p_subscribe_group == NULL )
		{
			p_subscribe_group = (struct MQSubscribeGroup *)malloc( sizeof(struct MQSubscribeGroup) ) ;
			if( p_subscribe_group == NULL )
			{
				ERRORLOG( "malloc sizeof(struct MQSubscribeGroup) failed , errno[%d]" , errno )
				MEMPACK_RESULT( break , MQSDK_ERROR_MALLOC , p_accepted_session->comm_data )
				break;
			}
			memset( p_subscribe_group , 0x00 , sizeof(struct MQSubscribeGroup) );
			strncpy( p_subscribe_group->subscribe_group_name , subscribe_group_name , subscribe_group_name_len );
			p_subscribe_group->subscribe_group_name_len = subscribe_group_name_len ;
			INIT_LIST_HEAD( & (p_subscribe_group->subscribe_session_list) );
			
			p_subscribe_group->p_store = p_store ;
			p_subscribe_group->msg_file = list_last_entry( & (p_store->msg_file_list) , struct MQFileMapping , file_list_node ) ;
			p_subscribe_group->p_topic = p_topic ;
			p_subscribe_group->topic_file = list_last_entry( & (p_topic->topic_file_list) , struct MQFileMapping , file_list_node ) ;
			p_subscribe_group->topic_fill_ptr = p_subscribe_group->topic_file->fill_ptr ;
			
			LinkSubscribeGroupNameTreeNode( p_topic , p_subscribe_group );
			INFOLOG( "add subscribe_group[%p][%s] to topic" , p_subscribe_group,p_subscribe_group->subscribe_group_name )
		}
		else
		{
			INFOLOG( "QuerySubscribeGroupTreeNode ok , subscribe_group[%p][%s]" , p_subscribe_group,p_subscribe_group->subscribe_group_name )
		}
		
		p_accepted_session->p_subscribe_group = p_subscribe_group ;
		p_accepted_session->long_pull_as_server_status = MQ_LONG_PULL_AS_SERVER_STATUS_LAST_SEND_AFTER_SILENT ;
		INFOLOG( "add accepted_session[%p] to subscribe_group[%p][%s]" , p_accepted_session , p_subscribe_group,p_subscribe_group->subscribe_group_name )
		list_add_tail( & (p_accepted_session->subscribe_group_subscribe_session_list_node) , & (p_subscribe_group->subscribe_session_list) );
		
		INFOLOG( "add accepted_session[%p] to msg_file[%p]->subscribe_msg_file_list" , p_accepted_session , p_subscribe_group->msg_file )
		list_add_tail( & (p_accepted_session->msg_file_subscribe_session_list_node) , & (p_subscribe_group->msg_file->subscribe_session_list) );
		INFOLOG( "add accepted_session[%p] to topic_file[%p]->subscribe_topic_file_list" , p_accepted_session , p_subscribe_group->topic_file )
		list_add_tail( & (p_accepted_session->topic_file_subscribe_session_list_node) , & (p_subscribe_group->topic_file->subscribe_session_list) );
		
		INFOLOG( "subscribe_name[%.*s] : store_name[%.*s] topic_name[%.*s] message" , subscribe_group_name_len,subscribe_group_name , store_name_len,store_name , topic_name_len,topic_name )
		
		MEMPACK_RESULT( break , MQSDK_SUCCESS , p_accepted_session->comm_data )
	}
	while(0);
	
	DEBUGHEXLOG( GetAppendBufferBase(p_accepted_session->comm_data) , GetAppendBufferLength(p_accepted_session->comm_data) , "pack ok , [%zu]bytes" , GetAppendBufferLength(p_accepted_session->comm_data) )
	
	return TCPMAIN_RETURN_WAITINGFOR_SENDING;
}

/*
CONSUME MESSAGE :
     CMQ
APP <---- BROKER
    ---->
     CMP

REQUEST MESSAGE PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|topic_name_len(1bytes)|topic(topic_name_len bytes)|tags_len(1bytes)|tags(tags_len bytes)|message_len(1bytes)|message(message_len bytes)|
|'0'|" CMQ"|0x07|"MYTOPIC"|0x06|"MYTAGS"|0x01|0x09|"MYMESSAGE"|

RESPONSE MESSAGE PROTOCOL :
|comm_version(1bytes)|process_code(4bytes)|result(2bytes)|topic_offset(8bytes)|
|'0'|" CMP"|0x0000|12345678|
*/

int app_ConsumeMessage( struct TcpdaemonServerEnvironment *p_env , struct MQAcceptedSession *p_accepted_session )
{
	uint16_t		result ;
	
	int			nret = 0 ;
	
	MEMUNPACK_RESULT( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , result )
	
	INFOLOG( "RECV CONSUME MESSAGE RESPONSE - result[%d]" , result )
	
	p_accepted_session->comm_status = TCPMAIN_RETURN_WAITINGFOR_RECEIVING ;
	p_accepted_session->p_subscribe_group->topic_fill_step = 0 ;
	p_accepted_session->long_pull_as_server_status = MQ_LONG_PULL_AS_SERVER_STATUS_SILENT ;
	
	nret = FetchTopicMessage( p_env , p_accepted_session->p_subscribe_group->p_topic ) ;
	if( nret < 0 )
	{
		FATALLOG( "FetchTopicMessage topic[%s] failed[%d]" , p_accepted_session->p_subscribe_group->p_topic->topic_name , nret )
		return TCPMAIN_RETURN_CLOSE;
	}
	else
	{
		INFOLOG( "FetchTopicMessage topic[%s] ok" , p_accepted_session->p_subscribe_group->p_topic->topic_name )
	}
	
	return p_accepted_session->comm_status;
}

/*
HEARTBEAT :
|comm_version(1bytes)|process_code(4bytes)|
|'0'|" HBQ"|

     HBQ
APP ----> BROKER
    <----
     HBQ
*/

int app_Heartbeat( struct TcpdaemonServerEnvironment *p_env , struct MQAcceptedSession *p_accepted_session )
{
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	
	int			nret = 0 ;
	
	if( p_accepted_session->long_pull_as_server_status != MQ_LONG_PULL_AS_SERVER_STATUS_SILENT )
	{
		DEBUGLOG( "long_pull_as_server_status[%d] != MQ_LONG_PULL_AS_SERVER_STATUS_SILENT[%d]" , p_accepted_session->long_pull_as_server_status , MQ_LONG_PULL_AS_SERVER_STATUS_SILENT )
		return TCPMAIN_RETURN_SILENCE;
	}
	
	ResetAppendBufferFillPtr( p_accepted_session->comm_data ) ;
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return TCPMAIN_RETURN_CLOSE , comm_version , p_accepted_session->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_HEARTBEAT_REQUEST ;
	MEMPACK_PROCESS_CODE( return TCPMAIN_RETURN_CLOSE , process_code , p_accepted_session->comm_data )
	
	p_accepted_session->long_pull_as_server_status = MQ_LONG_PULL_AS_SERVER_STATUS_FIRST_SEND_BEFORE_SILENT ;
	
	TDModifySessionOutputEvent( p_env , p_accepted_session->tcpdaemon_session );
	
	INFOLOG( "COMMIT SEND HEARTBEAT REQUEST" )
	
	return TCPMAIN_RETURN_WAITINGFOR_SENDING;
}

int app_HeartbeatResponse( struct TcpdaemonServerEnvironment *p_env , struct MQAcceptedSession *p_accepted_session )
{
	uint16_t		result ;
	
	int			nret = 0 ;
	
	MEMUNPACK_RESULT( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , result )
	
	INFOLOG( "RECV HEARTBEAT MESSAGE RESPONSE - result[%d]" , result )
	
	p_accepted_session->long_pull_as_server_status = MQ_LONG_PULL_AS_SERVER_STATUS_SILENT ;
	p_accepted_session->comm_status = TCPMAIN_RETURN_WAITINGFOR_RECEIVING ;
	
	nret = FetchTopicMessage( p_env , p_accepted_session->p_subscribe_group->p_topic ) ;
	if( nret < 0 )
	{
		FATALLOG( "FetchTopicMessage topic[%s] failed[%d]" , p_accepted_session->p_subscribe_group->p_topic->topic_name , nret )
		return TCPMAIN_RETURN_CLOSE;
	}
	else
	{
		INFOLOG( "FetchTopicMessage topic[%s] ok" , p_accepted_session->p_subscribe_group->p_topic->topic_name )
	}
	
	return p_accepted_session->comm_status;
}

/*
PRODUCE DELAY MESSAGE :

    PDMQ
APP ----> BROKER
    <----
     PDMP

|comm_version(1bytes)|process_code(4bytes)|delay_seconds(4bytes)|store_name_len(1bytes)|store_name(store_name_len bytes)|topic_name_len(1bytes)|topic(topic_name_len bytes)|tags_len(2bytes)|tags(tags_len bytes)|msg_data_len(4bytes)|msg_data(msg_data_len bytes)|
|'0'|"PDMQ"|0x05||0x07|"MYSTORE"|0x07|"MYTOPIC"|0x06|"MYTAGS"|0x09|"MYMSGDATA"|

|comm_version(1bytes)|process_code(4bytes)|result(2bytes)|delay_msg_offset(8bytes)|
|'0'|"PDMQ"|0x0000|12345678|

*/

int app_ProduceDelayMessage( struct TcpdaemonServerEnvironment *p_env , struct MQAcceptedSession *p_accepted_session )
{
	struct MQBroker			*broker = p_accepted_session->broker ;
	
	uint8_t				store_name_len ;
	char				*store_name = NULL ;
	uint8_t				topic_name_len ;
	char				*topic_name = NULL ;
	uint16_t			tags_len ;
	char				*tags = NULL ;
	uint32_t			msg_data_len ;
	char				*msg_data = NULL ;
	uint32_t			delay_seconds ;
	uint8_t				comm_version ;
	union MQProcessCode		process_code ;
	
	struct MQDelayMessage		*delay_msg = NULL ;
	
	struct MQStore			store ;
	struct MQStore			*p_store = NULL ;
	struct MQTopic			topic ;
	struct MQTopic			*p_topic = NULL ;
	struct MQFileMapping		*last_delay_msg_file = NULL ;
	size_t				total_delay_msg_data_unit_size ;
	ssize_t				wrote_len ;
	uint64_t			delay_msg_offset = 0 ;
	char				*rollback_fill_ptr = NULL ;
	
	struct MQFileMapping		*first_delay_msg_file = NULL ;
	
	int				nret = 0 ;
	
	MEMUNPACK_DELAY_SECONDS( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , delay_seconds )
	
	MEMUNPACK_STORE_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len )
	MEMUNPACK_STORE_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len , store_name )
	
	MEMUNPACK_TOPIC_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , topic_name_len )
	MEMUNPACK_TOPIC_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , topic_name_len , topic_name )
	
	MEMUNPACK_TAGS_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , tags_len )
	MEMUNPACK_TAGS( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , tags_len , tags )
	
	MEMUNPACK_MSG_DATA_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , msg_data_len )
	MEMUNPACK_MSG_DATA( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , msg_data_len , msg_data )
	
	INFOLOG( "RECV PRODUCE DELAY MESSAGE REQUEST - delay_seconds[%"PRIu32"] store_name[%.*s] topic_name[%.*s] tags[%.*s] msg_data[%.*s]" , delay_seconds , store_name_len,store_name , topic_name_len,topic_name , tags_len,tags , msg_data_len,msg_data )
	
	ResetAppendBufferFillPtr( p_accepted_session->comm_data ) ;
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return TCPMAIN_RETURN_CLOSE , comm_version , p_accepted_session->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_PRODUCEDELAYMESSAGE_RESPONSE ;
	MEMPACK_PROCESS_CODE( return TCPMAIN_RETURN_CLOSE , process_code , p_accepted_session->comm_data )
	
	do
	{
		memset( & store , 0x00 , sizeof(struct MQStore) );
		strncpy( store.store_name , store_name , store_name_len );
		store.store_name_len = store_name_len ;
		p_store = QueryStoreNameWithLenTreeNode( broker , & store ) ;
		if( p_store == NULL )
		{
			ERRORLOG( "store[%s] not exist" , store.store_name )
			MEMPACK_RESULT( break , MQSDK_ERROR_STORE_NOT_EXIST , p_accepted_session->comm_data )
			break;
		}
		else
		{
			DEBUGLOG( "store[%s] exist" , store.store_name )
		}
		
		memset( & topic , 0x00 , sizeof(struct MQTopic) );
		strncpy( topic.topic_name , topic_name , topic_name_len );
		topic.topic_name_len = topic_name_len ;
		p_topic = QueryTopicNameWithLenTreeNode( p_store , & topic ) ;
		if( p_topic == NULL )
		{
			ERRORLOG( "topic[%s] not exist" , topic.topic_name )
			MEMPACK_RESULT( break , MQSDK_ERROR_STORE_NOT_EXIST , p_accepted_session->comm_data )
			break;
		}
		else
		{
			DEBUGLOG( "topic[%s] exist" , topic.topic_name )
		}
		
		
		last_delay_msg_file = list_last_entry( & (p_store->delay_msg_file_list) , struct MQFileMapping , file_list_node ) ;
		total_delay_msg_data_unit_size = CalcDelayMessageDataUnitSize( topic_name_len , tags_len , msg_data_len ) ;
		DEBUGLOG( "last_delay_msg_file->fill_ptr[0x%X] + sizeof(uint64_t)[%zu] + total_delay_msg_data_unit_size[%zu] = [0x%X] > last_delay_msg_file->over_end_ptr[0x%X] ? [%d]" , last_delay_msg_file->fill_ptr , sizeof(uint64_t) , total_delay_msg_data_unit_size , last_delay_msg_file->fill_ptr+sizeof(uint64_t)+total_delay_msg_data_unit_size , last_delay_msg_file->over_end_ptr , last_delay_msg_file->fill_ptr+sizeof(uint64_t)+total_delay_msg_data_unit_size>last_delay_msg_file->over_end_ptr )
		if( last_delay_msg_file->fill_ptr + total_delay_msg_data_unit_size + sizeof(uint64_t) > last_delay_msg_file->over_end_ptr )
		{
			uint64_t	new_delay_msg_file_begin_offset = last_delay_msg_file->begin_offset + last_delay_msg_file->file_size ;
			
			*(uint64_t*)(last_delay_msg_file->fill_ptr) = hton64(UINT64_MAX) ;
			
			p_store->max_delay_msg_offset += last_delay_msg_file->over_end_ptr - last_delay_msg_file->fill_ptr ;
			DEBUGLOG( "set p_store->max_delay_msg_offset[%zu]" , p_store->max_delay_msg_offset )
			
			nret = CreateDelayMessageDataFile( p_accepted_session->comm_data , broker , p_store , new_delay_msg_file_begin_offset , & last_delay_msg_file , NULL ) ;
			if( nret )
			{
				ERRORLOG( "CreateDelayMessageDataFile failed , new_delay_msg_file_begin_offset[%"PRIu64"]" , new_delay_msg_file_begin_offset )
				free( delay_msg );
				MEMPACK_RESULT( break , MQSDK_ERROR_LOAD_DELAY_MSG_FILE , p_accepted_session->comm_data )
				break;
			}
			else
			{
				INFOLOG( "CreateDelayMessageDataFile ok , new_delay_msg_file_begin_offset[%"PRIu64"]" , new_delay_msg_file_begin_offset )
			}
		}
		
		rollback_fill_ptr = last_delay_msg_file->fill_ptr ;
		
		delay_msg_offset = last_delay_msg_file->fill_ptr-last_delay_msg_file->file_mmap + last_delay_msg_file->begin_offset ;
		INFOLOG( "before writing delay_msg_file , delay_msg_offset[%"PRIu64"] p_store->max_delay_msg_offset[%"PRIu64"]" , delay_msg_offset , p_store->max_delay_msg_offset )
		wrote_len = MemWrite_DelayMessageDataUnit( last_delay_msg_file->fill_ptr , delay_msg_offset , delay_seconds , MQ_DELAY_MSG_WAIT_TO_SEND , store_name_len , store_name , topic_name_len , topic_name , tags_len , tags , msg_data_len , msg_data ) ;
		DEBUGHEXLOG( last_delay_msg_file->fill_ptr , wrote_len , "wrote store_name[%.*s] topic_name[%.*s] tags[%.*s] msg_data[%.*s] to delay_msg_file[%s][%zu] in offset[%"PRIu64"] fill[%zu]bytes" , store_name_len,store_name , topic_name_len,topic_name , tags_len,tags , msg_data_len,msg_data , last_delay_msg_file->path_file_name , last_delay_msg_file->begin_offset , delay_msg_offset , wrote_len )
		
		delay_msg = CreateDelayMessage( p_env , p_accepted_session->comm_data , last_delay_msg_file , rollback_fill_ptr , delay_msg_offset , delay_seconds ) ;
		if( delay_msg == NULL )
		{
			ERRORLOG( "CreateDelayMessage failed" )
			MemWrite_DelayMessageDataUnit( rollback_fill_ptr , 0 , 0 , MQ_DELAY_MSG_ROLLBACK , 0 , NULL , 0 , NULL , 0 , NULL , 0 , NULL );
			break;
		}
		else
		{
			DEBUGLOG( "CreateDelayMessage ok" )
		}
		
		LinkDelayMessageOffsetTreeNode( p_store , delay_msg );
		LinkDelayMessageSendTimestampTreeNode( p_store , delay_msg );
		
		last_delay_msg_file->fill_ptr += wrote_len ;
		p_store->max_delay_msg_offset = wrote_len ;
		INFOLOG( "after writing delay_msg_file ,  delay_msg_offset[%"PRIu64"] p_store->max_delay_msg_offset[%"PRIu64"]" , delay_msg_offset , p_store->max_delay_msg_offset )
		
		rollback_fill_ptr = NULL ;
		
		MEMPACK_RESULT( break , MQSDK_SUCCESS , p_accepted_session->comm_data )
	}
	while(0);
	
	MEMPACK_DELAY_MSG_OFFSET( return TCPMAIN_RETURN_CLOSE , delay_msg_offset , p_accepted_session->comm_data )
	
	if( rollback_fill_ptr )
	{
		INFOLOG( "rollback fill ptr[%p]" , rollback_fill_ptr )
		last_delay_msg_file->fill_ptr = rollback_fill_ptr ;
	}
	
	DEBUGHEXLOG( GetAppendBufferBase(p_accepted_session->comm_data) , GetAppendBufferLength(p_accepted_session->comm_data) , "pack ok , [%zu]bytes" , GetAppendBufferLength(p_accepted_session->comm_data) )
	
	DEBUGLOG( "p_store->first_not_sent_delay_msg_file[%p] ->last_not_sent_delay_msg_file[%p]" , p_store->first_not_sent_delay_msg_file , p_store->last_not_sent_delay_msg_file )
	
	if( p_store )
	{
		DEBUGLOG( "p_store->delay_msg_file_list_nodes_count[%zu] > broker->delay_msg_file_list_nodes_soft_max[%zu] ?" , p_store->delay_msg_file_list_nodes_count , broker->delay_msg_file_list_nodes_soft_max )
		if( p_store->delay_msg_file_list_nodes_count > broker->delay_msg_file_list_nodes_soft_max )
		{
			first_delay_msg_file = list_first_entry( & (p_store->delay_msg_file_list) , struct MQFileMapping , file_list_node ) ;
			if( first_delay_msg_file )
			{
				DEBUGLOG( "list_first_entry delay_msg_file->path_file_name[%s]" , first_delay_msg_file->path_file_name )
				if( first_delay_msg_file != p_store->first_not_sent_delay_msg_file )
				{
					DEBUGLOG( "CleanDelayMessageDataFile" )
					CleanDelayMessageDataFile( broker , p_store , first_delay_msg_file );
				}
			}
		}
	}
	
	return TCPMAIN_RETURN_WAITINGFOR_SENDING;
}

/*
CANCEL DELAY MESSAGE :

    CDMQ
APP ----> BROKER
    <----
     CDMP

|comm_version(1bytes)|process_code(4bytes)|store_name_len(1bytes)|store_name(store_name_len bytes)|delay_msg_offset(8bytes)|
|'0'|"PDMQ"|0x07|"MYSTORE"|123|

|comm_version(1bytes)|result(2bytes)|process_code(4bytes)|
|'0'|"PDMP"|0x0000|
*/

int app_CancelDelayMessage( struct TcpdaemonServerEnvironment *p_env , struct MQAcceptedSession *p_accepted_session )
{
	struct MQBroker			*broker = p_accepted_session->broker ;
	
	uint8_t				store_name_len ;
	char				*store_name = NULL ;
	uint64_t			delay_msg_offset ;
	uint8_t				comm_version ;
	union MQProcessCode		process_code ;
	
	struct MQStore			store ;
	struct MQStore			*p_store = NULL ;
	struct MQFileMapping		*delay_msg_file = NULL ;
	ssize_t				read_len ;
	char				*delay_msg_ptr = NULL ;
	uint64_t			check_delay_msg_offset ;
	char				is_sent ;
	struct MQDelayMessage		delay_msg ;
	struct MQDelayMessage		*p_delay_msg = NULL ;
	
	int				nret = 0 ;
	
	MEMUNPACK_STORE_NAME_LEN( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len )
	MEMUNPACK_STORE_NAME( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , store_name_len , store_name )
	
	MEMUNPACK_DELAY_MSG_OFFSET( return TCPMAIN_RETURN_CLOSE , p_accepted_session->comm_data , delay_msg_offset )
	
	INFOLOG( "RECV PRODUCE DELAY MESSAGE REQUEST - delay_msg_offset[%"PRIu64"] store_name[%.*s]" , delay_msg_offset , store_name_len,store_name )
	
	ResetAppendBufferFillPtr( p_accepted_session->comm_data ) ;
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return TCPMAIN_RETURN_CLOSE , comm_version , p_accepted_session->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_CANCELDELAYMESSAGE_RESPONSE ;
	MEMPACK_PROCESS_CODE( return TCPMAIN_RETURN_CLOSE , process_code , p_accepted_session->comm_data )
	
	do
	{
		memset( & store , 0x00 , sizeof(struct MQStore) );
		strncpy( store.store_name , store_name , store_name_len );
		store.store_name_len = store_name_len ;
		p_store = QueryStoreNameWithLenTreeNode( broker , & store ) ;
		if( p_store == NULL )
		{
			ERRORLOG( "store[%s] not exist" , store.store_name )
			MEMPACK_RESULT( break , MQSDK_ERROR_STORE_NOT_EXIST , p_accepted_session->comm_data )
			break;
		}
		else
		{
			DEBUGLOG( "store[%s] exist" , store.store_name )
		}
		
		delay_msg_file = list_last_entry( & (p_store->delay_msg_file_list) , struct MQFileMapping , file_list_node ) ;
		if( delay_msg_offset > delay_msg_file->begin_offset+delay_msg_file->file_size-1 )
		{
			ERRORLOG( "delay_msg_offset[%"PRIu64"] is too big" , delay_msg_offset )
			MEMPACK_RESULT( break , MQSDK_ERROR_DELAY_MSG_OFFSET_IS_TOO_BIG , p_accepted_session->comm_data )
			break;
		}
		for( ; ; )
		{
			if( delay_msg_file->begin_offset <= delay_msg_offset && delay_msg_offset <= delay_msg_file->begin_offset+delay_msg_file->file_size-1 )
			{
				break;
			}
			
			delay_msg_file = list_prev_entry( delay_msg_file , struct MQFileMapping , file_list_node ) ;
			if( & (delay_msg_file->file_list_node) == & (p_store->msg_file_list) )
			{
				FATALLOG( "delay_msg_offset[%"PRIu64"] is cleaned" , delay_msg_offset )
				MEMPACK_RESULT( break , MQSDK_ERROR_DELAY_MSG_OFFSET_CLEANED , p_accepted_session->comm_data )
				break;
			}
		}
		
		delay_msg_ptr = delay_msg_file->file_mmap+(delay_msg_offset-delay_msg_file->begin_offset) ;
		read_len = MemRead_DelayMessageDataUnit( delay_msg_ptr , & check_delay_msg_offset , NULL , & is_sent , NULL , NULL , NULL , NULL , NULL , NULL , NULL , NULL ) ;
		if( check_delay_msg_offset != delay_msg_offset )
		{
			ERRORLOG( "delay_msg_offset is invalid , delay_msg_offset[%"PRIu64"]" , delay_msg_offset )
			MEMPACK_RESULT( break , MQSDK_ERROR_DELAY_MSG_OFFSET_INVALID , p_accepted_session->comm_data )
			break;
		}
		if( read_len == 0 )
		{
			ERRORLOG( "delay_msg is not exist , delay_msg_offset[%"PRIu64"]" , delay_msg_offset )
			MEMPACK_RESULT( break , MQSDK_ERROR_DELAY_MSG_NOT_EXIST_IN_STORE , p_accepted_session->comm_data )
			break;
		}
		
		if( is_sent != MQ_DELAY_MSG_WAIT_TO_SEND )
		{
			ERRORLOG( "delay_msg is_sent status[%c] is unexpected" , is_sent )
			MEMPACK_RESULT( break , MQSDK_ERROR_DELAY_MSG_SENT_FLAG_INVALID , p_accepted_session->comm_data )
			break;
		}
		
		memset( & delay_msg , 0x00 , sizeof(struct MQDelayMessage) );
		delay_msg.delay_msg_offset = delay_msg_offset ;
		p_delay_msg = QueryDelayMessageOffsetTreeNode( p_store , & delay_msg ) ;
		if( p_delay_msg == NULL )
		{
			FATALLOG( "QueryDelayMessageOffsetTreeNode return NULL , but continue processing" )
		}
		else
		{
			INFOLOG( "UnlinkDelayMessageOffsetTreeNode delay_msg_offset[%"PRIu64"]" , delay_msg_offset )
			UnlinkDelayMessageOffsetTreeNode( p_store , p_delay_msg );
			INFOLOG( "UnlinkDelayMessageSendTimestampTreeNode delay_msg_offset[%"PRIu64"]" , delay_msg_offset )
			UnlinkDelayMessageSendTimestampTreeNode( p_store , p_delay_msg );
			
			INFOLOG( "DestroyDelayMessage delay_msg_offset[%"PRIu64"]" , delay_msg_offset )
			DestroyDelayMessage( p_env , p_delay_msg );
		}
		
		INFOLOG( "delay_msg_offset[%"PRIu64"] set to cancel" )
		MemSet_DelayMessageDataUnit_is_sent( delay_msg_ptr , MQ_DELAY_MSG_CANCEL );
		
		MEMPACK_RESULT( break , MQSDK_SUCCESS , p_accepted_session->comm_data )
	}
	while(0);
	
	return TCPMAIN_RETURN_WAITINGFOR_SENDING;
}

